package org.example;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.ScalarFunction;

import java.util.HashMap;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
import java.util.Iterator;

/**
 * 自定义UDF函数：将JSON字符串转换为Map
 * 使用Flink内置的Jackson库进行JSON解析，支持完整的JSON结构转换
 * 
 * 此函数将JSON结构转换为Map<String, String>，特点：
 * - 保持JSON的原始结构，包括数组类型
 * - 支持任意深度的嵌套结构
 * - 处理对象、数组和基本值类型三种节点结构
 * - 所有值都转换为JSON字符串形式
 * 
 * 使用场景：
 * 1. 处理复杂的CDC消息，保持原始数据结构
 * 2. 在SQL中方便地访问JSON数据的任意层级
 * 3. 将JSON转为结构化数据便于分析和处理
 * 
 * 示例：
 * 假设有JSON字符串:
 * ```
 * {
 *   "user": {
 *     "id": "1001",
 *     "name": "张三",
 *     "address": {
 *       "city": "北京",
 *       "zip": "100000"
 *     }
 *   },
 *   "orders": [
 *     {
 *       "id": "A001",
 *       "amount": 100
 *     },
 *     {
 *       "id": "A002",
 *       "amount": 200
 *     }
 *   ]
 * }
 * ```
 * 
 * 转换后的Map将包含：
 * | 键                 | 值（JSON字符串）                    |
 * |-------------------|-----------------------------------|
 * | user              | {"id":"1001","name":"张三",...}    |
 * | user.id           | "1001"                            |
 * | user.name         | "张三"                            |
 * | user.address      | {"city":"北京","zip":"100000"}     |
 * | user.address.city | "北京"                            |
 * | user.address.zip  | "100000"                          |
 * | orders            | [{"id":"A001","amount":100},...]   |
 * | orders[0]         | {"id":"A001","amount":100}         |
 * | orders[0].id      | "A001"                            |
 * | orders[0].amount  | "100"                             |
 * | orders[1]         | {"id":"A002","amount":200}         |
 * | orders[1].id      | "A002"                            |
 * | orders[1].amount  | "200"                             |
 */
@FunctionHint(output = @DataTypeHint("MAP<STRING, STRING>"))
public class JsonToMapFunction extends ScalarFunction {
    
    private static final ObjectMapper objectMapper = new ObjectMapper();
    
    /**
     * 将JSON字符串转换为Map结构
     * 
     * 处理原则:
     * 1. 保持JSON的原始结构
     * 2. 数组类型保持为JSON数组字符串
     * 3. 对象类型保持为JSON对象字符串
     * 4. 空值和异常情况返回空Map
     * 
     * @param jsonStr JSON字符串
     * @return 包含完整结构的Map<String,String>对象，所有值都是JSON字符串
     */
    public Map<String, String> eval(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return new HashMap<>();
        }
        
        try {
            // 使用Jackson解析JSON字符串为JsonNode
            JsonNode jsonNode = objectMapper.readTree(jsonStr);
            
            // 创建结果Map
            Map<String, String> resultMap = new HashMap<>();
            
            // 处理节点
            processNode(jsonNode, "", resultMap);
            
            return resultMap;
        } catch (Exception e) {
            // 发生异常时返回空Map
            System.out.println("JSON解析异常: " + e.getMessage() + ", JSON: " + jsonStr);
            return new HashMap<>();
        }
    }
    
    /**
     * 处理JSON节点，将值转换为JSON字符串
     * 
     * @param node 当前处理的JSON节点
     * @param prefix 当前节点的前缀路径
     * @param resultMap 存储结果的Map对象
     */
    private void processNode(JsonNode node, String prefix, Map<String, String> resultMap) {
        try {
            if (node.isObject()) {
                // 处理对象节点
                String jsonStr = objectMapper.writeValueAsString(node);
                resultMap.put(prefix, jsonStr);
                
                // 递归处理对象的所有字段
                ObjectNode objectNode = (ObjectNode) node;
                Iterator<Map.Entry<String, JsonNode>> fields = objectNode.fields();
                while (fields.hasNext()) {
                    Map.Entry<String, JsonNode> field = fields.next();
                    String key = field.getKey();
                    JsonNode value = field.getValue();
                    
                    String newKey = prefix.isEmpty() ? key : prefix + "." + key;
                    processNode(value, newKey, resultMap);
                }
            } else if (node.isArray()) {
                // 处理数组节点
                String jsonStr = objectMapper.writeValueAsString(node);
                resultMap.put(prefix, jsonStr);
                
                // 递归处理数组的所有元素
                ArrayNode arrayNode = (ArrayNode) node;
                for (int i = 0; i < arrayNode.size(); i++) {
                    JsonNode element = arrayNode.get(i);
                    String newKey = prefix + "[" + i + "]";
                    processNode(element, newKey, resultMap);
                }
            } else if (node.isNull()) {
                resultMap.put(prefix, "null");
            } else if (node.isBoolean()) {
                resultMap.put(prefix, String.valueOf(node.asBoolean()));
            } else if (node.isInt()) {
                resultMap.put(prefix, String.valueOf(node.asInt()));
            } else if (node.isLong()) {
                resultMap.put(prefix, String.valueOf(node.asLong()));
            } else if (node.isDouble()) {
                resultMap.put(prefix, String.valueOf(node.asDouble()));
            } else {
                // 字符串类型
                resultMap.put(prefix, node.asText());
            }
        } catch (Exception e) {
            System.out.println("处理节点异常: " + e.getMessage() + ", 节点: " + node);
        }
    }
} 